package com.jianying.day07;

import com.jianying.beans.WaterSensor;
import com.jianying.func.WaterSensorMapFunction;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * ClassName: Flink05_TopN_1
 * Package: com.atguigu.day07
 * Description:
 *
 * @Author: tubage
 * @Create: 2024/4/8 14:13
 * @Version: 1.0
 */
public class Flink05_TopN_1 {
    public static void main(String[] args) throws Exception {
        // TODO top N 案例 统计最近10秒钟内出现次数最多的两个水位，并且每5秒钟更新一次
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> wsDS = env.socketTextStream("localhost", 8888)
                .map(new WaterSensorMapFunction());

        // TODO 指定watermark
        SingleOutputStreamOperator<WaterSensor> withWaterMarkDS = wsDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<WaterSensor>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<WaterSensor>() {
                                    @Override
                                    public long extractTimestamp(WaterSensor waterSensor, long l) {
                                        return waterSensor.getTs() * 1000;
                                    }
                                }
                        )
        );

        // TODO 事件时间窗口  不用KeyBy，开窗后并行度变为1
        AllWindowedStream<WaterSensor, TimeWindow> windowDS = withWaterMarkDS.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)));

        // TODO 全量处理
        windowDS.process(
                new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>() {
                    // TODO map（水位值，计数）、转为list 、排序
                    @Override
                    public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        Map<Integer, Integer> countMap = new HashMap<>();
                        for (WaterSensor element : elements) {
                            int wsValue = element.getVc();
                            countMap.put(wsValue, countMap.getOrDefault(wsValue, 0) + 1);
                        }

                        List<Map.Entry<Integer, Integer>> sortedEntries = new ArrayList<>(countMap.entrySet());
                        sortedEntries.sort((e1, e2) -> e2.getValue().compareTo(e1.getValue()));

                        int topN = Math.min(2, sortedEntries.size());
                        StringBuilder outStr = new StringBuilder();
                        outStr.append("================================\n");

                        for (int i = 0; i < topN; i++) {
                            Map.Entry<Integer, Integer> entry = sortedEntries.get(i);
                            // out.collect(new Tuple2<>(entry.getKey(), entry.getValue()) );
                            outStr.append("Top" + (i + 1) + "\n");
                            outStr.append("vc=" + entry.getKey() + "\n");
                            outStr.append("count=" + entry.getValue() + "\n");
                            outStr.append("窗口结束时间=" + DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss.SSS") + "\n");
                            outStr.append("================================\n");
                        }
                        out.collect(outStr.toString());
                    }

                }
        ).print();

        env.execute();
    }
}
